package com.jsm.tf.net;

import com.google.common.collect.Maps;
import com.jsm.tf.net.message.Message;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.Promise;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
public class RpcSyncPromise {
    private static final Map<Long, Promise<Object>> promises = Maps.newConcurrentMap();
    private int DEFAULT_TIME_OUT_MILLIS = 5000;

    public <T> T sync(Class<T> clazz, long seq, EventExecutor eventExecutor) {
        DefaultPromise<Object> promise = new DefaultPromise<Object>(eventExecutor);
        promises.put(seq, promise);
        try {
            promise.await(DEFAULT_TIME_OUT_MILLIS);
            if (promise.isSuccess()) {
                return (T) promise.getNow();
            } else {
                throw new RuntimeException(promise.cause());
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            promises.remove(seq);
        }
    }

    public void ack(long seq, Message message) {
        Promise<Object> promise = promises.get(seq);
        if (promise != null) {
            promise.setSuccess(message);
        }
    }
}
